/*
 * RMTransactionsFacadeImpl.java    ver   23/nov/2009
 *
 * Copyright (c) 2009 mentalsmash.org
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * The use of the Apache License does not indicate that this project is
 * affiliated with the Apache Software Foundation.
 */
package org.mentalsmash.tazio.transactions.management.rm;


import java.util.HashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.mentalsmash.documentation.Concurrency;
import org.mentalsmash.tazio.commons.identifiers.OID;
import org.mentalsmash.tazio.commons.timer.MonitorTimePerformances;
import org.mentalsmash.tazio.net.ril.RemoteObject;
import org.mentalsmash.tazio.transactions.TransactionStatus;
import org.mentalsmash.tazio.transactions.TransactionalSystemRuntimeException;
import org.mentalsmash.tazio.transactions.identifiers.TID;
import org.mentalsmash.tazio.transactions.management.StatusResponse;
import org.mentalsmash.tazio.transactions.management.TazioTransaction;
import org.mentalsmash.tazio.transactions.management.TxStateObserver;
import org.mentalsmash.tazio.transactions.objects.RootSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.inject.Inject;
import com.google.inject.Singleton;
import com.google.inject.internal.Maps;
import com.google.inject.internal.Preconditions;
import com.google.inject.name.Named;

/**
 * @version ver 23/nov/2009
 * @author Andrea Reale <andrea.reale@gmail.com>
 * 
 */
@Concurrency(concurrentAccess = true, specification = "The RMTransactionsFacadeImpl is a singleton and therfore is accessed concurrently"
	+ "by multiple threads. Nevertheless it has only final fields, so no particular synchronization attention is needed."
	+ "The lock named 'statusChangedLock' is used only to make threads wait for status changes.")
@Singleton
class RMTransactionsFacadeImpl implements
	RMTransactionsFacade, TxStateObserver {

    // RMI Addition
    private static final long serialVersionUID = -5923682893608651176L;
    // endof RMI Addition

    // in milliseconds
    private final int waitTimeoutLength;

    private final static Logger log = LoggerFactory
	    .getLogger(RMTransactionsFacadeImpl.class);

    private final RMTransactionManager tm;
    private final GroupTransactionsFacade group;
    private final RootSet rs;

    private final Lock statusChangedLock = new ReentrantLock();

    // Following variables are for monitoring
    private final Object monitorLock = new Object();
    private int numBeginRequests = 0;
    private int numWattRequests = 0;
    private int numUnknownWatt = 0;
    private int numRSRequests = 0;
    private int numCommitRequests = 0;

    private final HashMap<TID, Condition> waitingObjects = Maps.newHashMap();

    @Inject
    RMTransactionsFacadeImpl(final RMTransactionManager tm,
	    @Named("GroupProxy") final GroupTransactionsFacade groupFacade,
	    final RootSet rs, @Named("RMFacadeWaitTimeout") int waitTimeout) {
	this.tm = tm;
	Preconditions.checkArgument(groupFacade instanceof RemoteObject, "BAAHAHHA");
	this.group = groupFacade;
	this.rs = rs;
	this.waitTimeoutLength = waitTimeout;
	
	
	log.debug("RMTransactionsFacadeImpl created OID="+debugOID);
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.rm.RMTransactionsFacade#beginTransaction
     * ()
     */
    @Override
    public long beginTransaction() {
	log.debug("Sequence number request");
	// just return the current sequence number that is the number of the
	// last
	// committed transaction
	synchronized (monitorLock) {
	    numBeginRequests++;
	}
	try {
	    return tm.getSeqNoCtr();
	} finally {
	    log.debug("Sequence number responded");
	}
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.rm.RMTransactionsFacade#commitTransaction
     * (org.mentalsmash.tazio.transactions.control.TazioTransaction)
     */
    @MonitorTimePerformances
    @Override
    public long commitTransaction(final TazioTransaction transaction) {

	final TID tid = transaction.getTransactionId();
	log.debug("Commit request for {}", tid);
	synchronized (monitorLock) {
	    numCommitRequests++;
	}

	// creating monitor on the map

	if (waitingObjects.containsKey(tid)) {
	    throw new TransactionalSystemRuntimeException(
		    "Already received a pending request for tid" + tid);
	} else {
	    waitingObjects.put(tid, statusChangedLock.newCondition());
	    log.debug("OID="+debugOID+"; Created condition for {}", tid);
	}

	// send a broadcast message ordering to commit the transaction (the
	// message is broadcasted also to self)
	group.doCommit(tm.getManagerName(), transaction); // async request

	// after that call transaction becomes meaningless since the broadcast
	// sends parameters by copy and not by reference

	statusChangedLock.lock();
	try {
	    Condition condition = waitingObjects.get(tid);
	    TransactionStatus status = tm.getTransactionsLog().getStatusByTID(
		    tid);

	    // may lock if the transaction gets never commmitted
	    while ((status == null || status.isActive())) {

		boolean timeoutElapsed = false;

		try {

		    // waits for the condition of the transaction having its
		    // status changed (await return false if the timeout has
		    // elapsed)
		    log.debug("{} is wating for condition", tid);
		    timeoutElapsed = !condition.await(this.waitTimeoutLength,
			    TimeUnit.MILLISECONDS);
		} catch (final InterruptedException e) {
		    log.error(
			    "Thread awaiting for commitment outcome was interrupted."
				    + "Returning rollback to client", e);
		    return -1;
		}

		if (timeoutElapsed) {
		    log
			    .error(
				    "Timeout elapsed but no outcome about transaction {}",
				    tid);
		    return -1;
		}

		log.debug("{} woke up", tid);
		status = tm.getTransactionsLog().getStatusByTID(tid);
	    }
	    log.debug("{} Received status: {}", tid, status);

	    if (status == TransactionStatus.ROLLED_BACK) {
		return -1;
	    } else if (status == TransactionStatus.COMMITTED) {
		TazioTransaction tr = tm.getTransactionsLog().getByTID(tid);
		return tr.getSeqNo();
//		return tm.getTransactionsLog().getByTID(tid).getSeqNo();
	    } else { // TransactionStatus.NO_TRANSACTION
		log.error("The log returned a NO_TRANSACTION Status "
			+ "for transaction {}." + "This should not happen."
			+ "Returning rollback to client");
		return -1;
	    }

	} finally {

	    waitingObjects.remove(tid);
	    statusChangedLock.unlock();
	    log.debug("OID="+debugOID+"; Condition object {} removed", tid);
	}

    }

//    /*
//     * (non-Javadoc)
//     * 
//     * @see
//     * org.mentalsmash.tazio.transactions.rm.RMTransactionsFacade#getRootSet()
//     */
//    @Override
//    public RootSet getRootSet() {
//	synchronized (monitorLock) {
//	    numRSRequests++;
//	}
//	return this.rs;
//    }

    /*
     * (non-Javadoc)
     * 
     * @see org.mentalsmash.tazio.transactions.rm.RMTransactionsFacade#lock()
     */
    @Override
    public void lock() {
	// TODO Auto-generated method stub

    }

    /*
     * (non-Javadoc)
     * 
     * @see org.mentalsmash.tazio.transactions.rm.RMTransactionsFacade#unlock()
     */
    @Override
    public void unlock() {
	// TODO Auto-generated method stub

    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.rm.RMTransactionsFacade#whatAboutThisTx
     * (org.mentalsmash.tazio.transactions.model.identifiers.TID)
     */
    @Override
    public StatusResponse whatAboutThisTx(final TID tid) {
	final StatusResponse response;
	final TransactionStatus status = tm.getTransactionsLog()
		.getStatusByTID(tid);

	synchronized (monitorLock) {
	    numWattRequests++;
	}

	if (status == null || status == TransactionStatus.NO_TRANSACTION) {
	    if (status == TransactionStatus.NO_TRANSACTION) {
		log.warn("NO_TRANSACTION status for tx {}", tid);
	    }

	    response = new StatusResponseImpl(
		    StatusResponse.StatusResponseEnum.UNKNOWN, null);
	    synchronized (monitorLock) {
		numUnknownWatt++;
	    }

	} else if (status == TransactionStatus.COMMITTED) {
	    final long seqno = tm.getTransactionsLog().getByTID(tid).getSeqNo();
	    response = new StatusResponseImpl(
		    StatusResponse.StatusResponseEnum.COMMITTED, seqno);
	} else if (status == TransactionStatus.ROLLED_BACK) {
	    response = new StatusResponseImpl(
		    StatusResponse.StatusResponseEnum.ROLLED_BACK, null);
	} else {
	    response = new StatusResponseImpl(
		    StatusResponse.StatusResponseEnum.ACTIVE, null);
	}

	return response;

    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.control.TxStateObserver#statusChanged
     * (org.mentalsmash.tazio.transactions.user.TransactionStatus,
     * org.mentalsmash.tazio.transactions.user.TransactionStatus,
     * org.mentalsmash.tazio.transactions.model.identifiers.TID)
     */
    @Override
    public void statusChanged(final TransactionStatus from,
	    final TransactionStatus to, final TID tid) {

	if (to.equals(TransactionStatus.COMMITTED)
		|| to.equals(TransactionStatus.ROLLED_BACK)) {
	    // retrieve the condition associated to the tid
	    statusChangedLock.lock();
	    try {
		Condition cond = waitingObjects.get(tid);
		if (cond != null) {
		    cond.signal();
		} else {
		    log
			    .debug("OID="+debugOID+"; Condition was null ... maybe the thread did not wait?");
		}
	    } finally {
		statusChangedLock.unlock();
	    }
	}

    }
    
    private OID debugOID = OID.generateOID();

    /**
     * @see RMFacadeMonitorMBean
     * @return the numBeginRequests
     */
    protected int getNumBeginRequests() {
	synchronized (monitorLock) {
	    return numBeginRequests;
	}
    }

    /**
     * @see RMFacadeMonitorMBean
     * @return the numWattRequests
     */
    protected int getNumWattRequests() {
	synchronized (monitorLock) {
	    return numWattRequests;
	}
    }

    /**
     * @see RMFacadeMonitorMBean
     * @return the numUnknownWatt
     */
    protected int getNumUnknownWatt() {
	synchronized (monitorLock) {
	    return numUnknownWatt;
	}
    }

    /**
     * @see RMFacadeMonitorMBean
     * @return the numRSRequests
     */
    protected int getNumRSRequests() {
	synchronized (monitorLock) {
	    return numRSRequests;
	}
    }

    /**
     * @see RMFacadeMonitorMBean
     * @return the numCommitRequests
     */
    protected int getNumCommitRequests() {
	synchronized (monitorLock) {
	    return numCommitRequests;
	}
    }
}

